package cn.com.ytst.data.integration.components.reader.file;

import cn.com.ytst.data.integration.core.channel.PipleChannel;
import cn.com.ytst.data.integration.core.plugin.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.google.common.io.Files;
import com.google.common.io.LineProcessor;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @ClassName FileReader
 * @Deacription
 * @Author Administrator
 * @Date 2020/1/11 0011 17:10
 * @Version 1.0
 **/
public class FileReader extends Reader {

    private Logger logger = LoggerFactory.getLogger(this.getClass());


    String filepath;
    String filename;
    String spilt;
    String encode;

    @Override
    public void init() {
        Configuration configuration = this.getPluginJobConf();
        this.filepath = configuration.getString(FileReaderKey.FILE_READER_PATH_TAG);
        this.filename = configuration.getString(FileReaderKey.FILE_READER_NAME_TAG);
        this.spilt = configuration.getString(FileReaderKey.FILE_READER_SPLIT_TAG, FileReaderKey.DEFAULT_VALUE);
        this.encode = configuration.getString(FileReaderKey.FILE_READER_ENCODE_TAG,"utf-8");
    }

    @Override
    public void startReader() {
      //需要考虑如果读完了，就要把他迁移走，以防万一再次被读取
        try {
            File source =  new File(filepath+File.separator+filename);
            LineProcessor<String> lineProcessor =  new FileLineProcess(this.getPipleChannel());
            Files.readLines(source, Charset.forName("utf-8"), lineProcessor);
            logger.info("当前处理数据:{}条",lineProcessor.getResult());
            //进行文件的迁移
            File backfile = new File(this.filepath+File.separator+".bak");
            if(!backfile.exists()){
                FileUtils.forceMkdir(backfile);
            }
            FileUtils.moveFile(source,new File(backfile.getPath()+File.separator+filename));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void destroy() {
       
    }



    protected static class FileLineProcess implements LineProcessor<String>{


        private AtomicLong number ;

        private PipleChannel channel;

        public FileLineProcess(PipleChannel channel){
            this.channel = channel;
            this. number = new AtomicLong(0);
        }

        /**
         * This method will be called once for each line.
         *
         * @param line the line read from the input, without delimiter
         * @return true to continue processing, false to stop
         */
        @Override
        public boolean processLine(String line) throws IOException {
            //表示处理每一行
            channel.doPush(line);
            this. number.incrementAndGet();//计数器加一
            return true;
        }

        /**
         * Return the result of processing all the lines.
         */
        @Override
        public String getResult() {
            return String.valueOf(number.get());
        }
    }

}
